package top.easyblog.core.mq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.util.DigestUtils;
import top.easyblog.common.constant.Constants;
import top.easyblog.core.annotation.CostTimeTracker;
import top.easyblog.core.support.lock.DistributedLock;
import top.easyblog.core.support.lock.RedisDistributeLockFactory;
import top.easyblog.core.util.AssertUtil;
import top.easyblog.core.util.IdGenerator;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

import static top.easyblog.common.constant.RedisDistributeKeyConstants.REDIS_MQ_CONSUMER_LOCK_PREFIX;

/**
 * @author: frank.huang
 * @date: 2023-12-29 22:44
 */
@Slf4j
public abstract class AbstractRedisMessageConsumer<T> implements MessageListener {



    @CostTimeTracker
    @Override
    public void onMessage(Message message, byte[] bytes) {
        String requestId = MDC.get(Constants.REQUEST_ID);
        if (StringUtils.isBlank(requestId)) {
            MDC.put(Constants.REQUEST_ID, IdGenerator.getRequestId());
        }

        String channel = new String(bytes);
        log.info("onMessage --> 消息通道是：{}", channel);
        DistributedLock lock = null;
        try {
            T deserialize = deserialize(message);
            log.info("反序列化的结果:{}", deserialize);
            if (deserialize == null) {
                return;
            }

            //互斥消费，使用分布式来实现互斥
            String md5DigestAsHex = DigestUtils.md5DigestAsHex(deserialize.toString().getBytes(StandardCharsets.UTF_8));
            RedisDistributeLockFactory distributeLockFactory = getRedisDistributeLockFactory();
            AssertUtil.notNull(distributeLockFactory, "Required param distributeLockFactory can not be null");
            lock = distributeLockFactory.getNonWaitingDistributedLock(String.format(REDIS_MQ_CONSUMER_LOCK_PREFIX, md5DigestAsHex));
            if (lock.lock()) {
                doTask(deserialize);
                log.info("处理redis消息完成");
            } else {
                log.info("其他服务处理中");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("处理redis消息异常：", e);
        } finally {
            if (Objects.nonNull(lock)) {
                lock.unlock();
            }

            MDC.clear();
        }
    }

    /**
     * 反序列消息内容
     *
     * @param message
     * @return
     */
    protected abstract T deserialize(Message message);

    /**
     * 实现互斥消费
     */
    protected abstract RedisDistributeLockFactory getRedisDistributeLockFactory();

    /**
     * 处理任务
     *
     * @param deserialize
     */
    protected abstract void doTask(T deserialize);


    /**
     * 监听的消息Topic
     *
     * @return
     */
    public abstract String getTopic();

    /**
     * 是否启用此消费者
     *
     * @return
     */
    public abstract boolean isEnabled();


}
